package com.meteor.raft.masterelection;

import com.alibaba.fastjson.JSON;
import com.meteor.raft.constant.CommonConstant;
import com.meteor.raft.core.RaftPeerManage;
import com.meteor.raft.misc.HttpClient;
import com.meteor.raft.misc.UrlUtil;
import com.meteor.raft.misc.WebUtils;
import com.meteor.raft.persistent.RaftPeer;
import com.meteor.raft.util.NetUtils;
import com.ning.http.client.AsyncCompletionHandler;
import com.ning.http.client.Response;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.net.HttpURLConnection;
import java.util.HashMap;
import java.util.Map;

/**
 * @author gang.wang1903@xiaobao100.cn
 * @Date: 2019/7/16 9:09 AM
 */
@Service
@Slf4j
public class ElectionServiceImpl implements ElectionService{


    @Autowired
    NetUtils netUtils;

    @Autowired
    RaftPeerManage raftPeerManage;

    @Override
    public void sendVote() {

        RaftPeer local = raftPeerManage.getPeers().get(netUtils.localServer());
        log.info("leader timeout, start voting,leader: {}, term: {}",
                JSON.toJSONString(raftPeerManage.getLeader()), local.term);

        raftPeerManage.reset();

        local.term.incrementAndGet();
        local.voteFor = local.ip;
        local.state = RaftPeer.State.CANDIDATE;

        Map<String, String> params = new HashMap<String, String>(1);
        params.put("vote", JSON.toJSONString(local));
        for (final String server : raftPeerManage.allServersIncludeMyself()) {
            final String url = UrlUtil.buildURL(server, CommonConstant.API_VOTE);
            try {
             HttpClient.HttpResult result =  HttpClient.httpPost(url, null, params,"utf-8");
             log.info("received approve from peer: {}", JSON.toJSONString(result.content));
             RaftPeer peer =  JSON.parseObject(result.content, RaftPeer.class);
             raftPeerManage.decideLeader(peer);
            } catch (Exception e) {
                log.warn("error while sending vote to server: {}", server);
            }
        }
    }

    @Override
    public RaftPeer receivedVote(RaftPeer remote) {
        if (!raftPeerManage.getPeers().containsKey(remote.getIp())) {
            throw new IllegalStateException("can not find peer: " + remote.ip);
        }
        RaftPeer local = raftPeerManage.getPeers().get(netUtils.localServer());
        if (remote.term.get() <= local.term.get()) {
            String msg = "received illegitimate vote" + ", voter-term:" + remote.term + ", vote-term:" + local.term;
            log.info(msg);
            if (StringUtils.isEmpty(local.voteFor)) {
                local.voteFor = local.ip;
            }
            return local;
        }
        if(!remote.getIp().equals(local.getIp())) {
            local.resetLeaderDue();
            local.state = RaftPeer.State.FOLLOWER;
            local.voteFor = remote.ip;
            local.term.set(remote.term.get());
        }

        log.info("vote {} as leader, term: {}", remote.ip, remote.term);

        return local;
    }
}
